/**
 * File: SimpleController.java
 * Created by: mhaimel
 * Created on: 6 Jul 2009
 * CVS:  $Id: SimpleController.java,v 1.2 2009/10/09 14:45:45 mhaimel Exp $
 */
package uk.ac.ebi.curtain.controller;

import java.io.Serializable;
import java.util.Collection;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import uk.ac.ebi.curtain.CurtainContext;
import uk.ac.ebi.curtain.configuration.CurtainConfig;
import uk.ac.ebi.curtain.management.IController;
import uk.ac.ebi.curtain.management.IControllerReport;
import uk.ac.ebi.curtain.management.IInterceptor;
import uk.ac.ebi.curtain.management.IProcessable;
import uk.ac.ebi.curtain.management.InterceptorInfo;

import com.google.inject.Inject;

/**
 * @author mhaimel
 *
 */
public abstract class SimpleController<I extends Serializable, O extends Serializable> implements IController<I, O>, IProcessable {
	private static final int _DEFAULT_SLEEP_TIME = 1000;
	private final Log log = LogFactory.getLog(this.getClass());
	
	private final ConcurrentLinkedQueue<IInterceptor<O>> interceptorList = new ConcurrentLinkedQueue<IInterceptor<O>>();
	private final ConcurrentLinkedQueue<I> inputList = new ConcurrentLinkedQueue<I>();
	private final ConcurrentLinkedQueue<InterceptorInfo> interList = new ConcurrentLinkedQueue<InterceptorInfo>();
	private final AtomicReference<String> queue = new AtomicReference<String>();
	private final AtomicInteger parallel = new AtomicInteger(0);
	private final AtomicBoolean shutdown = new AtomicBoolean(false);
	private final AtomicBoolean forceShutdown = new AtomicBoolean(false);

	private final AtomicReference<Thread> shutdownThread = new AtomicReference<Thread>();
	private final AtomicReference<CurtainContext> context = new AtomicReference<CurtainContext>();
	private AtomicReference<CurtainConfig> config = new AtomicReference<CurtainConfig>(); 

	public void addInterceptor(IInterceptor<O> interceptor) {
		getInterceptorList().add(interceptor);
	}

	public void setParallel(Integer parallel) {
		this.parallel.set(parallel);
	}
	
	protected Integer getParallel() {
		return parallel.get();
	}

	public void setQueueName(String queueName) {
		this.queue.set(queueName);
	}
	
	protected String getQueueName(){
		return queue.get();
	}
	
	private I getNextJob() {
		synchronized (inputList) {
			return inputList.poll();	
		}
	}

	private InterceptorInfo getNextInfo() {
		synchronized (interList) {
			return interList.poll();
		}
	}

	public void process() {
		while(hasMoreJobs()){
			submit(processData(getNextJob(), getNextInfo()));
		}
	}


	public IControllerReport call() throws Exception {
		registerShutdownhook();
		try{
			while(keepProcessing()){
				process();
				sleep();
			}		
			getLog().debug("Finished Processing job queue!");
			doShutdown();
			return getReturnCall();
		} finally{
			removeShutdownhook();
		}
	}
	
	protected void submit(O output) {
		if(null != output){
			for(IInterceptor<O> inter : getInterceptorList()){
				inter.handle(output, buildInterceptorInfo());
			}
		}
	}

	protected InterceptorInfo buildInterceptorInfo() {
		return new InterceptorInfo(getQueueName());
	}

	private boolean hasMoreJobs() {
		return !isForcedShutdown() && !inputList.isEmpty() && !Thread.currentThread().isInterrupted();
	}

	protected abstract O processData(I i, InterceptorInfo info);

	private void removeShutdownhook() {
		Runtime.getRuntime().removeShutdownHook(getShutdownThread());
	}

	protected IControllerReport getReturnCall(){
		return new SimpleReport("Finished with " + getQueueName());
	}

	public boolean keepProcessing() {
		return (!shutdown.get() || hasMoreData()) && !isForcedShutdown() && !Thread.currentThread().isInterrupted();
	}

	private boolean hasMoreData() {
		return !inputList.isEmpty();
	}

	protected void sleep() {
		if(!isForcedShutdown()){
			try {
				Thread.sleep(sleepTime());
			} catch (InterruptedException e) {
				getLog().warn("Thread sleep error");
				Thread.currentThread().interrupt();
			}
		}
	}

	private long sleepTime() {
		return _DEFAULT_SLEEP_TIME;
	}

	protected void doShutdown() {
		// do nothing
		forwardShutdown(isForcedShutdown(), buildInterceptorInfo());
	}

	protected boolean isForcedShutdown() {
		return forceShutdown.get();
	}

	private void registerShutdownhook() {
		Runtime.getRuntime().addShutdownHook(getShutdownThread());		
	}
	
	private Thread getShutdownThread() {
		if (null == shutdownThread.get()){
			Thread s = new Thread() {
			    public void run() { 
			    	shutDown(true, buildInterceptorInfo());
			    }
			};
			shutdownThread.compareAndSet(null, s);
		}
		return shutdownThread.get();
	}

	public void handle(I input, InterceptorInfo info) {
		synchronized (inputList) {			
			inputList.add(input);
		}
		synchronized (interList) {	
			interList.add(info);
		}
	}
	
	private Collection<IInterceptor<O>> getInterceptorList() {
		return interceptorList;
	}

	public void shutDown(boolean force, InterceptorInfo info) {
		if(force){
			this.forceShutdown.set(true);
			this.shutdown.set(true);
			forwardShutdown(force, info);
		} else{
			this.shutdown.set(true);
		}
	}
	
	private void forwardShutdown(boolean force, InterceptorInfo info){
		for(IInterceptor<O> inter : getInterceptorList()){
			inter.shutDown(force,info);
		}	
	}
	
	protected Log getLog() {
		return log;
	}
	
	@Inject
	public void setContext(CurtainContext context) {
		this.context.set(context);
	}

	protected CurtainContext getContext() {
		return context.get();
	}

	/**
	 * @return the config
	 */
	public CurtainConfig getConfig() {
		return config.get();
	}

	/**
	 * @param config the config to set
	 */
	@Inject
	public void setConfig(CurtainConfig config) {
		this.config.set(config);
	}
	
	
}
